package com.yikekong.emq;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
import com.yikekong.dto.DeviceInfoDTO;
import com.yikekong.service.AlarmService;
import com.yikekong.service.DeviceService;
import com.yikekong.service.QuotaService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 信息回调
 */
@Component
@Slf4j
public class EmqMsgProcess implements MqttCallback {

    @Autowired
    private EmqClient emqClient;
    @Autowired
    private QuotaService quotaService;
    @Autowired
    private AlarmService alarmServicel;
    @Autowired
    private DeviceService deviceService;
    /**
     * 丢失连接后再次连接和订阅
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.info("连接丢失，重新连接");
        emqClient.connect();
        quotaService.getAllSubject().forEach(s ->{
            try {
                emqClient.subscribe("$queue/"+s);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        });
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        String payload = new String(mqttMessage.getPayload());
        System.out.println("接收到数据："+payload);

        //json转map
        ObjectMapper mapper = new ObjectMapper();
        Map payloadMap = mapper.readValue(payload, Map.class);

        //解析报文
        DeviceInfoDTO deviceInfoDTO = quotaService.analysis(topic, payloadMap);
        if (deviceInfoDTO != null){

            //报警判断，封装报警
            deviceInfoDTO = alarmServicel.verifyDeviceInfo(deviceInfoDTO);
            //保存数据
            deviceService.saveDeviceInfo(deviceInfoDTO.getDevice());
            //保存数据到influxdb
            quotaService.saveQuotaToInflux(deviceInfoDTO.getQuotaList());
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }
}
